package com.yy.job.core.run.worker;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.yy.job.common.enums.JobActionEnum;
import com.yy.job.common.enums.JobExecuteTypeEnum;
import com.yy.job.constant.JobConstant;
import com.yy.job.core.config.xml.ModuleXmlConfig;
import com.yy.job.core.domain.JobDomain;
import com.yy.job.core.domain.context.JobContext;
import com.yy.job.core.run.flow.JobRunFlow;

/**
 * yy-job的后台线程，一直保持运行状态<br>
 * 每个module下面根据jobWorkerSize来创建对应数量的JobWorker<br>
 * 
 * @author eleven.li
 *
 */
public class JobWorker implements Runnable {

	private Logger log = LoggerFactory.getLogger(JobWorker.class);

	/**
	 * 队列长度，暂定为200<br>
	 * 队列太长的话，可能会引起排队过长超过锁定时间，任务被重新加入队列重复执行
	 */
	private static final Integer MAX_WORKER_QUEUE_COUNT = Integer.valueOf(50);
	
	/**
	 * 阻塞队列
	 */
	private BlockingQueue<JobDomain> jobQueue = null;
	
	/**
	 * 是否执行中
	 */
	private boolean isRuning = false;
	
	/**
	 * jobWorker名称
	 */
	private String jobWorkerName;

	private String jobWorkerNameFormat = "WORKER(%s)-%s";

	private ModuleXmlConfig moduleJobConfig;

	public JobWorker(ModuleXmlConfig moduleJobConfig, int workIndex, String moduleBossName) {
		this.moduleJobConfig = moduleJobConfig;
		this.jobWorkerName = String.format(jobWorkerNameFormat, workIndex, moduleBossName);
		
		// 基于链表的无界队列，但是还是设置固定长度以免oom
		//this.jobQueue = new LinkedBlockingQueue<>(MAX_WORKER_QUEUE_COUNT.intValue());
		// 无界队列。因为存在JobExecuteTypeEnum.PRIORITY的执行方式。此时需要突破队列上限。
		this.jobQueue = new LinkedBlockingQueue<>();
	}

	public synchronized void stop() {
		if (!this.isRuning) {
			return;
		}
		this.isRuning = false;
	}

	public synchronized void start() {
		if (this.isRuning) {
			return;
		}
		log.info("JobWoker name={} is starting ... ", this.jobWorkerName);
		
		this.isRuning = true;
		(new Thread(this)).start();
		log.info("JobWoker name={} started success", this.jobWorkerName);
	}

	@Override
	public void run() {
		while (this.isRuning) {
			try {
				// 从队列里面获取任务
				JobDomain jobDomain = this.jobQueue.take();
				log.info("take job from queue, param={}", JSON.toJSONString(jobDomain));
				if (jobDomain == null) {
					continue;
				}
				
				// 执行job相关逻辑
				JobRunFlow jobManager = new JobRunFlow(moduleJobConfig);
				JobContext jobContext = new JobContext(jobDomain, 
					String.format(JobConstant.FORMAT_OPERATOR, this.getJobWorkerName(),jobDomain.getRunCount()), JobActionEnum.AUTO);
				jobManager.flow(jobContext);
			}
			catch(Exception ex) {
				log.error("JobWorker run failed", ex);
			}
		}
	}
	
	/**
	 * 将job放到队列里面
	 * 
	 * @param job
	 * @return
	 */
	public boolean putJob(JobDomain job) {
		boolean isSuccess = false;
		try {
			// 如果是高优先级的，则支持任何时候插入
			if(JobExecuteTypeEnum.PRIORITY.equals(job.getJobExecuteType())) {
				// 尝试将元素插入队列，如果队列已满，那么会阻塞8秒直到有空间插入
				//isSuccess = this.jobQueue.offer(job, 8L, TimeUnit.SECONDS);
				isSuccess = this.jobQueue.offer(job);
			}
			else {
				if(this.jobQueue.size() > MAX_WORKER_QUEUE_COUNT) {
					return false;
				}
				isSuccess = this.jobQueue.offer(job, 8L, TimeUnit.SECONDS);
			}
		} catch (Exception e) {
			log.error("JobWorker offer failed", e);
			log.error("JobWorker offer failed, job={}" + JSON.toJSONString(job));
		}
		log.info("put job to queue, param={}, result={}", JSON.toJSONString(job), isSuccess);
		return isSuccess;
	}

	public String getJobWorkerName() {
		return jobWorkerName;
	}

}
